package com.pisolution.scheduler.remote.handler;

import com.pisolution.scheduler.core.tuple.Tuple2;
import com.pisolution.scheduler.remote.command.Command;
import com.pisolution.scheduler.remote.command.CommandType;
import com.pisolution.scheduler.remote.processor.NettyRequestProcessor;

import io.netty.channel.Channel;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;

/**
 * 服务端请求处理适配器 负责转发请求到适配的处理器.
 *
 * @author sora
 * @version 0.1.0
 */
@ChannelHandler.Sharable
public final class NettyServerHandler extends ChannelInboundHandlerAdapter {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyServerHandler.class);

    // -------------------------------------------------------------------------------------------------

    /** 以二元组的方式存储所有注册到服务端的处理器. */
    private final ConcurrentHashMap<CommandType, Tuple2<NettyRequestProcessor, ExecutorService>> processors;

    /** 用于创建处理器执行的环境. */
    private final ExecutorService executor;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public NettyServerHandler(final ExecutorService executor) {
        this.processors = new ConcurrentHashMap<>();
        this.executor = executor;
    }

    // -------------------------------------------------------------------------------------------------
    //  处理器注册
    // -------------------------------------------------------------------------------------------------

    /**
     * 注册给定的处理器以处理特定类型的请求.
     *
     * @param type 该处理器所用于处理的请求类型.
     * @param processor 待注册的处理器.
     * @param executor 用于创建执行请求时的执行环境 如果为空则使用服务端默认提供的实例.
     */
    public void register(final CommandType type, final NettyRequestProcessor processor, final ExecutorService executor) {
        this.processors.putIfAbsent(type, new Tuple2<>(processor, executor == null ? this.executor : executor));
    }

    // -------------------------------------------------------------------------------------------------
    //  错误处理与请求路由
    // -------------------------------------------------------------------------------------------------

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.error("Caught exception when processing the command received from the client.", cause);

        ctx.channel().close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();

            return;
        }

        super.userEventTriggered(ctx, evt);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        ctx.channel().close();
    }

    @SuppressWarnings("DuplicatedCode")
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 1. 转换消息对象为请求命令对象.
        final Channel channel = ctx.channel();
        final Command command = (Command) msg;

        // 2. 尝试寻找能够处理该请求命令的处理器.
        final Tuple2<NettyRequestProcessor, ExecutorService> processor = this.processors.get(command.getType());

        // 3. 确认是否支持该命令的处理.
        if (processor == null) {
            LOGGER.warn("The command of type {} currently not supported yet.", command.getType());

            return;
        }

        // 4. 使用指定的运行环境与处理器处理该请求命令.
        try {
            processor.field2.submit(() -> {
                try {
                    processor.field1.process(channel, command);
                } catch (Throwable t) {
                    LOGGER.error("Could not process the command {}.", command, t);
                }
            });
        } catch (RejectedExecutionException e) {
            LOGGER.warn(
                "The thread pool is full, discard the command {} received from {}.",
                command,
                ((InetSocketAddress) channel.remoteAddress()).getAddress().getHostAddress()
            );
        }
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
        // 1. 从上下文中提取出信道信息与配置.
        final Channel channel = ctx.channel();
        final ChannelConfig config = channel.config();

        // 2. 记录日志并更改自动读取配置开关.
        if (channel.isWritable()) {
            if (LOGGER.isWarnEnabled()) {
                LOGGER.warn(
                    "The channel {} is writeable, to low water {}.",
                    channel,
                    config.getWriteBufferLowWaterMark()
                );
            }

            config.setAutoRead(true);
            return;
        }

        if (LOGGER.isWarnEnabled()) {
            LOGGER.warn(
                "The channel {} is not writeable, over high water {}.",
                channel,
                config.getWriteBufferHighWaterMark()
            );
        }

        config.setAutoRead(false);
    }
}
